package com.liyu.rabbit.roll;

import com.liyu.rabbit.Producer;
import com.liyu.rabbit.factory.RabbitFactory;
import com.rabbitmq.client.Channel;

import java.io.IOException;
import java.util.concurrent.*;

/**
 * @author 李誉
 * @date 2022-05-15 15:50
 * @description 手动应答消费者
 */
public class ManualConsumer {

    public static void main(String[] args) throws IOException, TimeoutException {
        /**
         * 新建线程池
         *  corePoolSize:核心线程池大小
         *  maximumPoolSize:最大核心线程池大小
         *  keepAliveTime:空闲线程存活时间
         *  unit:时间单位
         *  workQueue:阻塞队列
         *  threadFactory:线程工厂：创建线程的，一般不用动
         *  handler:拒绝策略
         */
        ExecutorService executorService = new ThreadPoolExecutor(
                2,
                3,
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(2),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy()
        );


        for (int i=0; i < 2; i++) {
            int index = i;
            executorService.execute( () -> {
                try {
                    Channel newChannel = RabbitFactory.createNewChannel();
                    newChannel.basicConsume(Producer.QUEUE_NAME, false, (consumerTag, message) -> {
                        System.out.println("线程 -"+ index + "- 消费一条消息" + consumerTag + "：" + new String(message.getBody()));
                        if (index % 2 == 0) {
//                            try {
//                                Thread.sleep(20000);
//                            } catch (InterruptedException e) {
//                                e.printStackTrace();
//                            }
                            // 手动应答
                            // 参数：消息标签，是否批量，是否重新返回队列
                            newChannel.basicNack(message.getEnvelope().getDeliveryTag(), false, true);
                            System.out.println("线程 -"+ index + "- 消费消息" + consumerTag + "失败");
                            return;
                        }
                        // 手动应答，第二个参数控制是否批量应答
                        newChannel.basicAck(message.getEnvelope().getDeliveryTag(), false);
                        System.out.println("线程 -"+ index + "- 消费消息" + consumerTag + "成功");
                    }, (consumerTag) -> {
                        System.out.println("线程 -"+ index + "- 消费一条消息" + consumerTag + "失败：");
                    });
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }

            });
        }

    }
}
